package server;

import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.mqtt.*;

import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;

public class BrokerHandler extends SimpleChannelInboundHandler<MqttMessage> {
    private MqttVersion version;
    private String clientId;
    private String userName;
    private String brokerId;
    private boolean connected;
    private boolean cleanSession;
    private int keepAlive;
    private int keepAliveMax;
    private MqttPublishMessage willMessage;//遗嘱

    public BrokerHandler(int keepAliveMax) {
        this.keepAliveMax = keepAliveMax;
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
    }

    @Override
    @SuppressWarnings("ThrowableResultOfMethodCallIgnored")
    protected void channelRead0(ChannelHandlerContext ctx, MqttMessage msg) throws Exception {
        System.out.println(this.clientId+"  "+ msg.fixedHeader().messageType());
        //判断协议是否正确
        if (msg.decoderResult().isFailure()) {
            Throwable cause = msg.decoderResult().cause();
            if (cause instanceof MqttUnacceptableProtocolVersionException) {
                BrokerSessionHelper.sendMessage(
                        ctx,
                        MqttMessageFactory.newMessage(
                                new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
                                new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION, false),
                                null),
                        "INVALID",
                        null,
                        true);

            } else if (cause instanceof MqttIdentifierRejectedException) {

                BrokerSessionHelper.sendMessage(
                        ctx,
                        MqttMessageFactory.newMessage(
                                new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
                                new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED, false),
                                null),
                        "INVALID",
                        null,
                        true);
            }

            ctx.close();

            return;
        }

        switch (msg.fixedHeader().messageType()) {
            case CONNECT:
                onConnect(ctx, (MqttConnectMessage) msg);
                break;
            case PUBLISH:
                onPublish(ctx, (MqttPublishMessage) msg);
                break;
            case PUBACK:
                onPubAck(ctx, msg);
                break;
            case PUBREC:
                onPubRec(ctx, msg);
                break;
            case PUBREL:
                onPubRel(ctx, msg);
                break;
            case PUBCOMP:
                onPubComp(ctx, msg);
                break;
            case SUBSCRIBE:
                onSubscribe(ctx, (MqttSubscribeMessage) msg);
                break;
            case UNSUBSCRIBE:
                onUnsubscribe(ctx, (MqttUnsubscribeMessage) msg);
                break;
            case PINGREQ:
                onPingReq(ctx);
                break;
            case DISCONNECT:
                onDisconnect(ctx);
                break;
        }

    }

    //建立连接
    private void onConnect(ChannelHandlerContext ctx, MqttConnectMessage msg) {

        this.version = MqttVersion.fromProtocolNameAndLevel(msg.variableHeader().name(), (byte) msg.variableHeader().version());
        this.clientId = msg.payload().clientIdentifier();
        this.cleanSession = msg.variableHeader().isCleanSession();
        if (msg.variableHeader().keepAliveTimeSeconds() > 0 && msg.variableHeader().keepAliveTimeSeconds() <= this.keepAliveMax) {
            this.keepAlive = msg.variableHeader().keepAliveTimeSeconds();
        }

        System.out.println("版本:"+version);
        System.out.println("id:"+clientId);
        System.out.println("心跳时间:"+keepAlive);
        System.out.println("安全等级:"+msg.variableHeader().willQos());
        //MQTT 3.1之后可能存在为空的客户ID。所以要进行处理。如果客户ID是空，而且还在保存处理相关的信息。这样子是不行。
        //必须有客户ID我们才能存保相关信息。
        if (StringUtils.isBlank(this.clientId)) {
            if (!this.cleanSession) {

                BrokerSessionHelper.sendMessage(
                        ctx,
                        MqttMessageFactory.newMessage(
                                new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
                                new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED, false),
                                null),
                        "INVALID",
                        null,
                        true);

                ctx.close();

                return;

            } else {
                this.clientId =  java.util.UUID.randomUUID().toString();
            }
        }

        //如果发送多次连接就违规协议,关闭连接
        if (this.connected) {
            ctx.close();
            return;
        }

        boolean userNameFlag = msg.variableHeader().hasUserName();
        boolean passwordFlag = msg.variableHeader().hasPassword();
        this.userName = msg.payload().userName();

        String password = "" ;
        if( msg.payload().passwordInBytes() != null  && msg.payload().passwordInBytes().length > 0)
            password =   new String(msg.payload().passwordInBytes());
        System.out.println("username:"+userName);
        System.out.println("password:"+password);

        boolean mistake = false;

        //如果有用户名标示，那么就必须有密码标示。
        //当有用户名标的时候，用户不能为空。
        //当有密码标示的时候，密码不能为空。
        if (userNameFlag) {
            if (StringUtils.isBlank(this.userName))
                mistake = true;
        } else {
            if (StringUtils.isNotBlank(this.userName) || passwordFlag) mistake = true;
        }
        if (passwordFlag) {

            if (StringUtils.isBlank(password)) mistake = true;
        } else {
            if (StringUtils.isNotBlank(password)) mistake = true;
        }
        if (mistake) {
            BrokerSessionHelper.sendMessage(
                    ctx,
                    MqttMessageFactory.newMessage(
                            new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
                            new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD, false),
                            null),
                    this.clientId,
                    null,
                    true);
            ctx.close();
            return;
        }

        //回复连接
        BrokerSessionHelper.sendMessage(
                ctx,
                MqttMessageFactory.newMessage(
                        new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
                        new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_ACCEPTED, !this.cleanSession),
                        null),
                this.clientId,
                null,
                true);

        ChannelHandlerContext lastSession = BrokerSessionHelper.removeSession(this.clientId);
        if (lastSession != null) {
            lastSession.close();
        }

        //处理遗嘱
        if(msg.variableHeader().isWillFlag()){//是否有遗嘱
            String willTopic = msg.payload().willTopic();//遗嘱分发的主题
            String willMessage = "";
            if(msg.payload().willMessageInBytes() != null && msg.payload().willMessageInBytes().length > 0)
                willMessage =  new String(msg.payload().willMessageInBytes());
            if (msg.variableHeader().isWillFlag() && StringUtils.isNotEmpty(willTopic) && StringUtils.isNotEmpty(willMessage)) {
                this.willMessage = (MqttPublishMessage) MqttMessageFactory.newMessage(
                        new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.valueOf(msg.variableHeader().willQos()), msg.variableHeader().isWillRetain(), 0),
                        new MqttPublishVariableHeader(willTopic, 0),
                        Unpooled.wrappedBuffer(willMessage.getBytes())
                );
            }
        }


        this.connected = true;
        BrokerSessionHelper.saveSession(this.clientId, ctx);
    }

    //订阅
    private void onSubscribe(ChannelHandlerContext ctx, MqttSubscribeMessage msg) {

        if (!this.connected) {
            ctx.close();
            return;
        }
        int messageId = msg.variableHeader().messageId();

        //获取主题名字,为空直接关闭
        List<MqttTopicSubscription> requestSubscriptions = msg.payload().topicSubscriptions();
        for (MqttTopicSubscription subscription : requestSubscriptions) {
            if (StringUtils.isEmpty(subscription.topicName())) {
                ctx.close();
                return;
            }
        }

        List<Integer> grantedQosLevels   = new ArrayList<Integer>();

        requestSubscriptions.forEach(subscription -> {
            if (subscription.topicName().startsWith("$")) grantedQosLevels.add(MqttQoS.FAILURE.value());
            else grantedQosLevels.add(subscription.qualityOfService().value());
        });


        BrokerSessionHelper.sendMessage(
                ctx,
                MqttMessageFactory.newMessage(
                        new MqttFixedHeader(MqttMessageType.SUBACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
                        MqttMessageIdVariableHeader.from(messageId),
                        new MqttSubAckPayload(grantedQosLevels)),
                this.clientId,
                messageId,
                true);

        for (int i = 0; i < requestSubscriptions.size(); i++) {

            MqttQoS grantedQoS = MqttQoS.valueOf(grantedQosLevels.get(i));
            String topic = requestSubscriptions.get(i).topicName();

            //1。查看以前有没有订阅过相同的主题，如果有就替换。
            //2。查看有没有符合的保留信息，有发送
            //读者们自行去实现。是要用redis，还是要用sqllite自去实现。

        }
    }

    private void onUnsubscribe(ChannelHandlerContext ctx, MqttUnsubscribeMessage msg) {

    }

    //ping消息
    private void onPingReq(ChannelHandlerContext ctx) {
        BrokerSessionHelper.sendMessage(
                ctx,
                MqttMessageFactory.newMessage(
                        new MqttFixedHeader(MqttMessageType.PINGRESP, false, MqttQoS.AT_MOST_ONCE, false, 0),
                        new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED, false),
                        null),
                "INVALID",
                null,
                true);
    }

    private void onDisconnect(ChannelHandlerContext ctx) {

        if (!this.connected) {
            ctx.close();
            return;
        }
        BrokerSessionHelper.removeSession(this.clientId, ctx);
        this.willMessage = null;
        this.connected = false;
        ctx.close();

    }

    private void onPubComp(ChannelHandlerContext ctx, MqttMessage msg) {
        System.out.println(msg.toString());
    }

    private void onPubRel(ChannelHandlerContext ctx, MqttMessage msg) {
        System.out.println(msg.toString());
    }
    //qos2
    private void onPubRec(ChannelHandlerContext ctx, MqttMessage msg) {
        System.out.println(msg.toString());
    }

     //qos1消息确认，如果不回复，按协议会重发(并不会重发)
    private void onPubAck(ChannelHandlerContext ctx, MqttMessage msg) {
        System.out.println(msg.toString());
    }

    /**
     *     AT_MOST_ONCE(0),
     *     AT_LEAST_ONCE(1),
     *     EXACTLY_ONCE(2),
     * @param ctx
     * @param msg
     */
    private void onPublish(ChannelHandlerContext ctx, MqttPublishMessage msg) {
         MqttQoS qos = msg.fixedHeader().qosLevel();
        String result = msg.payload().toString(Charset.forName("GB2312"));
        System.out.println("接收到消息 id:"+this.clientId+ "消息安全等级:"+qos.value()+"  消息内容:"+result);
      //  MqttMessage message = new MqttMessage(new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.AT_MOST_ONCE, false, 0), null, ctx.alloc().buffer().writeBytes("你说啥?".getBytes()));

        MqttPublishMessage mqttPublishMessage = (MqttPublishMessage) MqttMessageFactory.newMessage(
                new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.AT_LEAST_ONCE, false, 0),
                new MqttPublishVariableHeader(msg.variableHeader().topicName(), msg.variableHeader().packetId()),
                Unpooled.wrappedBuffer(ctx.alloc().buffer().writeBytes("你说啥?".getBytes(Charset.forName("GB2312")))
                ));

        ctx.writeAndFlush(mqttPublishMessage);

    }
}
